package com.atguigu.flink.chapter07;

import com.atguigu.flink.chapter05.Source.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/20 14:06
 */
public class Flink27_State_StateBackend {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 状态后端配置
        // 1.配置 内存的状态后端
        MemoryStateBackend memoryStateBackend = new MemoryStateBackend();
        env.setStateBackend(memoryStateBackend);
        // 2.配置 文件系统的状态后端
        FsStateBackend fsStateBackend = new FsStateBackend("hdfs://hadoop1:8020/xxx/xxx/xx");
        env.setStateBackend(fsStateBackend);
        // 3.配置 RocksDB的状态后端
        RocksDBStateBackend rocksDBStateBackend = new RocksDBStateBackend("hdfs://hadoop1:8020/xxx/xx", true);
//        rocksDBStateBackend.setWriteBatchSize();  // 可以指定一些 RocksDB的参数
        env.setStateBackend(rocksDBStateBackend);

        // 一般配合checkpoint使用
        env.enableCheckpointing(200);


        env.execute();
    }

}
/*

 */